package com.atguigu.edu.realtime.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.utils.DimUtil;
import com.atguigu.edu.realtime.utils.DruidDSUtil;
import com.atguigu.edu.realtime.utils.RedisUtil;
import com.atguigu.edu.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;


public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取线程池
        threadPool = ThreadPoolUtil.getThreadPool();
    }
    
    public abstract String getTable();
    
    public abstract String getId(T input);
    
    public abstract void addDim(T input, JSONObject dim);
    
    /** 参数 1: 输入流中的数据类型 */
    /** 参数 2: 把要输出的数据, 放入到这个参数中 */
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        /** todo by me 步骤
         因为redis和phoenix的客户端没有支持异步，而使用flink异步IO的前提是要客户端要支持异步。
         采用的解决方案是【多线程 + 多客户端】

         先需要在open创建一个线程池，所以我们需要new的是一个RichAsyncFunction<>（两个泛型是输入和输出的类型） 才有open
         所有的数据处理都放在了线程池的线程里面，然后我们会用到redis和phoenix客户端，因为redis和phoenix客户端不支持异步模式，
         所以我们需要在线程池的每个子线程都创建一个redis和phoenix客户端，这里需要用到连接池保证效率。

         然后就是执行查询dim表的操作，先查redis 再查phoenix。注意这里需要用到表名和id，我们这里获取表名和id的方式就用抽象方法，
         让实现者自己传表名和id，再加一个addDim(input,dim)，来设置 维度信息
         input.setSkuName(dim.getString("SKU_NAME"));
         input.setSpuId(dim.getString("SPU_ID"));
         input.setTrademarkId(dim.getString("TM_ID"));
         input.setCategory3Id(dim.getString("CATEGORY3_ID"));

         一次只能补充一张表。其他表就再 AsyncDataStream.unorderedWait()。。。去实现，
         主要就是这里的这个AsyncFunction的区别 其他三个参数有待处理的流、超时时间、单位是一样的


         */
        //
        // 数据的所有处理,放入线程池
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                
                try {
                    Jedis redisClient = RedisUtil.getRedisClient();  // 一个线程获取一个客户端
                    DruidPooledConnection conn = DruidDSUtil.getPhoenixConn();
                    
                    JSONObject dim = DimUtil.readDim(redisClient, conn, getTable(), getId(input));
                    addDim(input, dim);  // 这张表的维度信息补充完毕. 子类去实现
                    
                    // 数据输出到后面的流中
                    resultFuture.complete(Collections.singletonList(input));
                    
                    redisClient.close();
                    conn.close();
                    
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        });
        
    }
}
